package yz.mr.MapJoin;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;

public class MapJoinMapper extends Mapper<LongWritable, Text, NullWritable,MapStudent> {
    HashMap<String,String> studentScore=new HashMap<String,String>();
    /**
     *setup()函数在Map之前执行了一次，那我们可以在setup函数中加载学生成绩数据到内存中进行保存
     * 之后再Map函数读取每一行的学生基本信息，在根据学生ID进行数据匹配关联
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void setup(Mapper<LongWritable, Text, NullWritable, MapStudent>.Context context) throws IOException, InterruptedException {
        //加载学生成绩数据，并形成一个HashMap   学生ID为Key  成绩作为value
        //context中保存了缓存路径
        URI[] cacheFiles = context.getCacheFiles();
        URI scoreFile = cacheFiles[0];

        //创建FileSystem  由于上下文对象保存了其配置类的对象，可以直接获取
        Configuration configuration = context.getConfiguration();
        FileSystem fileSystem = FileSystem.get(configuration);

        //创建输入流，并且按行读取
        FSDataInputStream fs = fileSystem.open(new Path(scoreFile));

        InputStreamReader inputStreamReader = new InputStreamReader(fs);
        BufferedReader bufferedReader = new BufferedReader(inputStreamReader);
        String line;
        while ((line=bufferedReader.readLine()) != null){
            String[] split = line.split("\t");
            String id = split[0];
            String score = split[1];
            //将读取到的数据读取到内存中
            studentScore.put(id,score);
        }

    }

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, NullWritable, MapStudent>.Context context) throws IOException, InterruptedException {
        String student = value.toString();
        String[] split = student.split("\t");
        String id = split[0];
        String[] split1 = split[1].split(",");
        //从对应的ID去取对应的Value
        String score = studentScore.get(id);
        MapStudent mapStudent = new MapStudent(id,split1[0],split1[1],split1[2],split1[3],Integer.parseInt(split1[4]));
        context.write(NullWritable.get(),mapStudent);

        System.out.println(student);
    }
}
